package com.sanerzone.smscenter.gateway.http.handler.listener;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.sanerzone.smscenter.gateway.base.GatewayFactory;
import com.sanerzone.smscenter.gateway.http.handler.listener.Task.Task;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.concurrent.*;

@Service
public class GateWayMTListener implements MessageListenerConcurrently
{
    static Logger logger = LoggerFactory.getLogger(GateWayMTListener.class);
    
    private GatewayFactory gatewayFactory;
    static  BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10000); //固定为10000的线程队列
    static  ThreadPoolExecutor pool;
    static {
        pool=new ThreadPoolExecutor( 50, 2000, 60, TimeUnit.SECONDS, queue, Executors.defaultThreadFactory(),
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        //拒绝策略 阻塞队列
                        try {
                            queue.put( r );
                        }catch (Exception e){
                            logger.info( "队列已满，新增消息阻塞中。" );
                        }
                    }
                } );
        logger.info( "发送线程池已经初始化成功！" );
    }
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context)
    {
        Task task = new Task( msgs,context,gatewayFactory );
//        CMPP30SMProxy
        pool.submit(task);//提交任务
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    
    public void setGatewayFactory(GatewayFactory gatewayFactory)
    {
        this.gatewayFactory = gatewayFactory;
    }

}
